package cn.lzd.kafka.cn.lzd.kafka.apache;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

/**
 * Created by lzd on 2018/8/1.
 */
public class ProducerMe extends Thread {

    static KafkaProducer<String, String> producer;

    public ProducerMe() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hdpn:9092,hdpd1:9092,hdpd2:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<>(props);
    }

    public static void main(String[] args) {
        ProducerMe me = new ProducerMe();
//        me.test1();
        me.test2();
    }

    public void test2() {
        String topic = "lzdtest";
        String key = "keyhello2";
        String val = "valhello2";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, val);
        producer.send(record, new DemoCallBack(System.currentTimeMillis(), key, val));
        producer.close();
        System.out.println("end ...");
    }

    public void test1() {
        String topic = "test";
        String key = "keyhello";
        String val = "valhello";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, val);
        producer.send(record);
        producer.close();
        System.out.println("end ...");
    }

}

class DemoCallBack implements Callback {

    private final long startTime;
    private final String key;
    private final String message;

    public DemoCallBack(long startTime, String key, String message) {
        this.startTime = startTime;
        this.key = key;
        this.message = message;
    }

    public void onCompletion(RecordMetadata metadata, Exception exception) {
        long elapsedTime = System.currentTimeMillis() - startTime;
        if (metadata != null) {
            System.out.println(
                    "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
                            "), " +
                            "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
        } else {
            exception.printStackTrace();
        }
    }
}